package com.carol.bigdata.utils

import org.apache.hadoop.hbase.filter._
import org.apache.hadoop.hbase.util.Bytes

object HBaseFilter {

    def getPatternFilter(pattern: String): RowFilter = {
        // val pattern = ".*_" + rowkeys.head + "$"
        val filter: RowFilter = new RowFilter(CompareFilter.CompareOp.valueOf("EQUAL"),
//            new RegexStringComparator(pattern))
        new SubstringComparator(pattern))
        filter
    }

    def getTimeFilter(dates: List[String]): List[RowFilter] = {
        var filterList = List[RowFilter]()
        if (dates.length == 1) {
            val filter: RowFilter = getPatternFilter(dates.head)
            filterList :+= filter
        }
        if (dates.length == 2) {
            val startFilter: RowFilter = new RowFilter(CompareFilter.CompareOp.GREATER_OR_EQUAL,
                new BinaryComparator(Bytes.toBytes(dates.head)))
            val stopFilter: RowFilter = new RowFilter(CompareFilter.CompareOp.LESS,
                new BinaryComparator(Bytes.toBytes(dates.last)))
            filterList :+= startFilter
            filterList :+= stopFilter
            //scan.setStartRow(Bytes.toBytes(rowkeys.head))
            //scan.setStopRow(Bytes.toBytes(rowkeys.last))
        }
        filterList
    }

    def getTimePatternFilter(rowkeys: List[String]): List[RowFilter] = {
        val filterList: List[RowFilter] = getTimeFilter(rowkeys.tail)
        filterList :+ getPatternFilter(rowkeys.head)
    }

    def getColumnFilter(filterCF: String,
                        filterCol: String,
                        filterValue: String): SingleColumnValueFilter = {
        //val tmpFilter: ValueFilter = new ValueFilter(CompareFilter.CompareOp.NOT_EQUAL, new BinaryComparator(Bytes.toBytes(filterValue)))
        //val filter : SkipFilter = new SkipFilter(tmpFilter)
        val filter: SingleColumnValueFilter = new SingleColumnValueFilter(
            Bytes.toBytes(filterCF),
            Bytes.toBytes(filterCol),
            CompareFilter.CompareOp.EQUAL,
            new BinaryComparator(Bytes.toBytes(filterValue))
        )
        filter
    }

}
